//Team 4 - Gold Coast
//Duy Kha Nguyen 598240 
//Rongkai Zhou 562946 
//Zhangyue Cheng 548692
//Binzan Liang 591911
//Wei Liu 572322
package twitter;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;

import com.twitter.hbc.ClientBuilder;
import com.twitter.hbc.core.Client;
import com.twitter.hbc.core.Constants;
import com.twitter.hbc.core.Hosts;
import com.twitter.hbc.core.HttpHosts;
import com.twitter.hbc.core.endpoint.Location;
import com.twitter.hbc.core.endpoint.Location.Coordinate;
import com.twitter.hbc.core.endpoint.StatusesFilterEndpoint;
import com.twitter.hbc.core.processor.StringDelimitedProcessor;
import com.twitter.hbc.httpclient.auth.Authentication;
import com.twitter.hbc.httpclient.auth.OAuth1;

public class CollectTweetTask implements Runnable {

    private static final Logger logger = Logger.getLogger(CollectTweetTask.class);

    private final BlockingQueue<String> tweetQueue;

    private final String consumerKey;

    private final String consumerSecret;

    private final String token;

    private final String tokenSecret;

    private final double swLatitude;

    private final double swLongitude;

    private final double neLatitude;

    private final double neLongitude;

    public CollectTweetTask(BlockingQueue<String> tweetQueue, String consumerKey, String consumerSecret, String token, String tokenSecret, double swLatitude,
            double swLongitude, double neLatitude, double neLongitude) {
        super();
        this.tweetQueue = tweetQueue;
        this.consumerKey = consumerKey;
        this.consumerSecret = consumerSecret;
        this.token = token;
        this.tokenSecret = tokenSecret;
        this.swLatitude = swLatitude;
        this.swLongitude = swLongitude;
        this.neLatitude = neLatitude;
        this.neLongitude = neLongitude;
    }

    public void run() {
        final Coordinate southWest = new Coordinate(swLatitude, swLongitude);
        final Coordinate northEast = new Coordinate(neLatitude, neLongitude);
        final List<Location> locations = new ArrayList<Location>();
        locations.add(new Location(southWest, northEast));

        final Authentication hosebirdAuth = new OAuth1(consumerKey, consumerSecret, token, tokenSecret);

        final Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
        final ClientBuilder builder = new ClientBuilder();
        builder.name("Hosebird-Client-01");
        builder.hosts(hosebirdHosts);
        builder.authentication(hosebirdAuth);
        builder.endpoint(new StatusesFilterEndpoint().locations(locations));
        builder.processor(new StringDelimitedProcessor(tweetQueue));

        Client hosebirdClient = null;
        try {
            hosebirdClient = builder.build();
            hosebirdClient.connect();
        } catch (Exception e) {
            logger.error("Error when invoking connect", e);
        }
    }

}
